package com.rainman.simpleredis.core.simplemq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.rainman.simpleredis.core.config.SimpleRedisConfig;
import com.rainman.simpleredis.core.distributed_lock.SimpleDistributedLock;
import com.rainman.simpleredis.core.scheduled.SimpleScheduled;
import com.rainman.simpleredis.core.simplemq.sch_worker.RealServerSchWorker;
import com.rainman.simpleredis.core.utils.DelayList;

import java.time.Duration;
import java.util.Set;


/**
 * 实时消息处理
 * 基于redis的消息发布订阅机制
 *
 */
public class SimpleMqRealServer {

    //原始数据头
    public  final String RealTimeHead;//SR0.Ori.RealMq:

    private final SimpleRedisConfig simpleRedisConfig;
    private final SimpleDistributedLock simpleDistributedLock;
    private final DelayList delayList;

    //死信头
    private  final String DHead;//SR0.Dead.RealMq:
    //分布式锁头数据
    public  final String DealLockHead="SimpleRedis.RealDealOne:";

    public SimpleMqRealServer(SimpleRedisConfig simpleRedisConfig
            , SimpleDistributedLock simpleDistributedLock
            , SimpleScheduled simpleScheduled){
        this.simpleRedisConfig = simpleRedisConfig;
        this.simpleDistributedLock=simpleDistributedLock;
        delayList=new DelayList();

        String keyHead=simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()+".";//SR0.
        RealTimeHead=keyHead+"Ori.RealMq:";//SR0.Ori.RealMq:
        DHead=keyHead+"Dead.RealMq:";//SR0.Dead.RealMq:

        simpleScheduled.addWorker(new RealServerSchWorker("SimpleMqReal.ScanData", this::scanData));

    }


    /**
     * 兜底扫描数据
     *
     */
    private void scanData() {

        //获取到所有原始信息的key
        Set<String> oriInfoKeys = simpleRedisConfig.getRedisInterface().keys(RealTimeHead+"*");//SR0.Ori.RealMq:*

        for(String key:oriInfoKeys){//单个key示例SR0.Ori.RealMq:类型值:目标id值

            //判断是否有处理锁
            Boolean isDealDLock = simpleDistributedLock.hasLock(DealLockHead+key);
            if(isDealDLock){
                continue;
            }

            try {
                Thread.sleep(100L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            //确认信息是否还存在
            Boolean aBoolean = simpleRedisConfig.getRedisInterface().hasKey(key);
            if(!aBoolean){
                continue;
            }

            //判断是否有处理锁
            isDealDLock = simpleDistributedLock.hasLock(DealLockHead+key);
            if(isDealDLock){
                continue;
            }


            //非死信 非延迟消息。则补充触发消息
            System.out.println("RealServer 兜底生效 可忽略"+key);

            String[] split = key.split(":");

            publishMQTig(split[1],split[2]);

        }

    }


    /**
     * 发布订阅触发单条消息处理
     *
     * @param key SR0.Ori.RealMq:类型值:目标id值
     */
    public void dealOne(String key){//SR0.Ori.RealMq:类型值:目标id值

        String[] split = key.split(":");

        String type=split[1];
        String id=split[2];

        //获取处理者
        SimpleMqWorkerInterface curWorker=null;
        for(SimpleMqWorkerInterface worker: simpleRedisConfig.getMqWorkerInterfaces()){
            if(worker.matchTargetType(type)){
                curWorker=worker;
            }
        }

        if(curWorker==null){//没有消费，放弃抢夺权限，
            return;
        }

        //处理逻辑
        String value = simpleRedisConfig.getRedisInterface().getValue(RealTimeHead + type + ":" + id);
        if(value==null){
            return;
        }

        JSONObject jsonObject = JSONUtil.parseObj(value);

        if(simpleRedisConfig.getVer()<jsonObject.getLong("ver")){
            return;
        }

        //抢之前看下自己状态,上次有异常状态的让。尽可能让其他消费者抢到
        boolean b = delayList.hasKey(key);
        if(b){
            try {
                Thread.sleep(20L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        String dLockKey=DealLockHead + key;//SimpleRedis.RealDealOne:SR0.Ori.RealMq:类型值:目标id值
        Boolean dLockAdd =simpleDistributedLock.dLockAdd(dLockKey,40000L);

        if(!dLockAdd){//没有抢夺到处理权
            return ;
        }
        if(b){
            delayList.remove(key);
        }

        //以下是处理

        WorkerReturn msg = new WorkerReturn(WorkerReturn.StateRetry);

        try{

            msg=curWorker.doWork(type, id, jsonObject.getStr("msg"));
        }finally {

            simpleDistributedLock.dLockRemove(dLockKey);//删除锁
            if (msg==null||WorkerReturn.StateSuc.equals(msg.getState())){

                simpleRedisConfig.getRedisInterface().delete(RealTimeHead+type+":"+id);//成功删除原始信息

            }else if (WorkerReturn.StateRetry.equals(msg.getState())){

                delayList.add(key); //下次兜底重试
            }else if (WorkerReturn.StateRetryImm.equals(msg.getState())){

                delayList.add(key);
                publishMQTig(type,id);//立即重试

            }else {//死信

                jsonObject.set("errMsg",msg.getFailMsg());
                publishMQDead(type,id,jsonObject.toString());//新增死信记录

                simpleRedisConfig.getRedisInterface().delete(RealTimeHead+type+":"+id);//删除原始信息,已转换为死信信息
            }
        }

    }



    public void publishMQOri(String targetType,  String targetId,String msgStr) {

        if(targetType.contains(":")){
            throw new RuntimeException("targetType不能包含非法字符 : ");
        }


        //转入触发时间，消息
        JSONObject jsonObject=new JSONObject();
        jsonObject.set("time",System.currentTimeMillis());
        jsonObject.set("msg",msgStr);
        jsonObject.set("para",true);
        jsonObject.set("ver", simpleRedisConfig.getVer());

        //插入原始数据
        simpleRedisConfig.getRedisInterface().set(RealTimeHead+targetType+":"+targetId,jsonObject.toString());
        //插入触发消息
        publishMQTig(targetType,targetId);

    }

    private void publishMQTig(String targetType,  String targetId) {
        simpleRedisConfig.getRedisInterface().convertAndSend(
                simpleRedisConfig.getKeyHead()+ simpleRedisConfig.getProduct()
                ,RealTimeHead+targetType+":"+targetId); //SR0.Ori.RealMq:targetType:targetId
    }

    protected void publishMQDead(String targetType,  String targetId,String valueStr) {//死信保留时间为15天
        simpleRedisConfig.getRedisInterface()
                .set(DHead+targetType+":"+targetId,valueStr, Duration.ofMillis(1000L*3600*24*15));

    }


}
